package com.lengxf.flink.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 无界数据读取
 * 事件驱动形
 *
 * @author Lengxf
 */
//运行是增加参数--add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/sun.nio.ch=ALL-UNNAMED
public class WordCountStreamUnboundedDemo {

    //算子并行度优先级
    //算子>代码：env >提交时指定> 配置文件
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        config.set(RestOptions.BIND_PORT, "8082");
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        //并行度
        //算子单独指定的优先级高于全局配置
        env.setParallelism(3);
        env.disableOperatorChaining();
        //从socket中读取数据
        //并进行处理
        env.socketTextStream("localhost", 8888)
                .flatMap(getStringTuple2FlatMapFunction())
                .setParallelism(4)
                //解决类型擦除的问题
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(v -> v.f0)
                .sum(1);
        env.execute();
    }

    private static FlatMapFunction<String, Tuple2<String, Integer>> getStringTuple2FlatMapFunction() {
        return (s, collector) -> {
            String[] split = s.split(" ");
            for (String word : split) {
                collector.collect(Tuple2.of(word, 1));
            }
        };
    }
}
